package com.atguigu.day03;

import com.atguigu.utils.IntSource;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Example2 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntSource())
                //r 是输入的参数，Tuple3.of() 构造3个一样的r
                .map(r -> Tuple3.of(r, r, r))
                // 类型擦除 需要提供map输出的类型注解
                .returns(Types.TUPLE(Types.INT,Types.INT,Types.INT))
                .keyBy(r -> true)
                .reduce(new ReduceFunction<Tuple3<Integer, Integer, Integer>>() {
                    @Override
                    public Tuple3<Integer, Integer, Integer> reduce(Tuple3<Integer, Integer, Integer> value, Tuple3<Integer, Integer, Integer> t1) throws Exception {
                        return Tuple3.of(
                                Math.min(value.f0, t1.f0),
                                Math.max(value.f0, t1.f0),
                                value.f0 + t1.f0
                        );
                    }
                })
                .print();







        env.execute();
    }
}
